package cn._51doit.day07;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @create: 2021-10-23 14:29
 * @author: 今晚打脑斧先森
 * @program: AdCount
 * @Description:
 *  对偏移量进行保存,但是不能保证当出现问题的时候,将当前的偏移量实时保存
 *  而是从上次保存的偏移量中继续读取数据
 *  保证偏移量的准确性,使用 synchronized (ctx.getCheckpointLock()){} 锁
 **/
public class AtLeastOnceFileSourceDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        env.enableCheckpointing(100000);
        DataStreamSource<String> res = env.addSource(new MySourceFunction("D:\\wokk\\operatorState"));

        DataStreamSource<String> lines = env.socketTextStream("doit01", 8888);
        SingleOutputStreamOperator<String> res2 = lines.map(new MapFunction<String, String>() {
            @Override
            public String map(String value) throws Exception {
                if (value.startsWith("error")) {
                    throw new RuntimeException("亲爱的今晚打脑斧先森,数据出错了哦");
                }
                return value;
            }
        });

        res.union(res2).print();    //因为自定义source和socketTextStream的流之间没有联系,socketTextStream出问
                                    //题,自定义source不会重启,所以union一起,就可以了,让他们之间有联系.
        env.execute();

    }
}
